package main

import (
	"fmt"
	"sync"
	"time"
)

func start(n int) <-chan string {
	out := make(chan string)
	go func() {
		defer close(out)
		for i := 0; i < n; i++ {
			time.Sleep(time.Millisecond * 100)
			out <- fmt.Sprint("配件", i)

		}
	}()
	return out
}

func build(in <-chan string) <-chan string {
	out := make(chan string)
	go func() {
		defer close(out)
		for c := range in {
			time.Sleep(time.Millisecond * 500)
			out <- fmt.Sprint("组装", "+", c)
		}
	}()
	return out
}

func end(in <-chan string) <-chan string {
	out := make(chan string)
	go func() {
		defer close(out)
		for c := range in {
			time.Sleep(time.Millisecond * 200)
			out <- fmt.Sprint("完成", "+", c)
		}
	}()
	return out
}

// merge 扇入函数（组件），可以合并多个管道到一个管道中
func merge(ins ...<-chan string) <-chan string {
	var wg sync.WaitGroup
	out := make(chan string)
	// 把一个通道中的数据发送到out中。
	p := func(in <-chan string) {
		defer wg.Done()
		for c := range in {
			out <- c
		}
	}
	// 扇入，启动多个goroutine，每一个处理一个管道
	for _, cs := range ins {
		wg.Add(1)
		go p(cs)
	}
	// 等待所有ins处理完后，关闭out管道
	go func() {
		wg.Wait()
		close(out)
	}()
	return out
}

func main() {
	begin := time.Now()
	s := start(10)
	b1 := build(s) // 扇出
	b2 := build(s)
	b3 := build(s)
	b4 := build(s)
	b5 := build(s)
	b6 := build(s)
	b := merge(b1, b2, b3, b4, b5, b6) // 扇入
	e1 := end(b)
	e2 := end(b)
	e3 := end(b)
	e := merge(e1, e2, e3)
	for item := range e {
		fmt.Println(item)
	}
	duration := time.Since(begin)
	fmt.Println("耗时:", duration)
}
